package org.example.component;

import com.alibaba.fastjson2.JSON;
import org.example.pojo.SeckillOrder;
import org.example.service.SeckillActivityService;
import org.example.service.SeckillOrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
public class BlockingQueueConsumer {
	
	@Autowired
	private RedisTemplate redisTemplate;
	@Autowired
	SeckillOrderService seckillOrderService;
	
	@Autowired
	SeckillActivityService seckillActivityService;

	private static final String QUEUE_KEY = "stock:queue";

	/**
	 * list队列获取消息
	 */
	@Async
	public void startConsuming()
	{
		System.out.println("开启消息队列...");
		while (true) {
			// 阻塞式弹出，超时时间设置为0表示无限等待			}
			String message = (String) redisTemplate.opsForList().leftPop(QUEUE_KEY, 0, TimeUnit.SECONDS);
			if (message != null) {
				processMessage(message);
			}
		}
	}

	// 处理消息逻辑
	private void processMessage(String message) {
		//保存数据库
		//创建订单
		//库存减少
//		System.out.println(message);
		SeckillOrder seckillOrder = (SeckillOrder)JSON.parseObject(message,SeckillOrder.class);
		seckillOrderService.save(seckillOrder);
		boolean update = seckillActivityService.update().setSql("total_stock = total_stock-1")//必须获取数据库里面的total_stock值-1
				.eq("coupon_id", seckillOrder.getCouponId())
				.gt("total_stock", 0)
				.update();//执行

	}

	/**
	 * redis stream获取消息
	 */
	@Async
	public void startConsumingRedisStream()
	{
		System.out.println("开启消息队列...");
		//创建组
		redisTemplate.opsForStream().createGroup("stream:order", ReadOffset.from("$"), "mygroup");
		
		while (true) {
			// 阻塞式弹出，超时时间设置为0表示无限等待			}
			// 执行XREADGROUP命令
			List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read(
					Consumer.from("mygroup", "user1"), // 消费者组和消费者名称
					StreamReadOptions.empty().count(1).block(Duration.ofMillis(5000)), // 选项：count 1, block 0
					StreamOffset.create("stream:order", ReadOffset.from(">")) // 流名称和读取位置">"
			);
			
			if (records.size() > 0) {
				process(records.get(0));
			}
		}
	}

	private void process(MapRecord mapRecord) 
	{
		//map结构
		Map value = (Map) mapRecord.getValue();
//		System.out.println(value);
		//获取JSON
		Object o = value.get("stream:order");
//		System.out.println(o);
		SeckillOrder seckillOrder = JSON.parseObject(o.toString(), SeckillOrder.class);
		
		//生成订单
		seckillOrderService.save(seckillOrder);
		//库存减一
		boolean update = seckillActivityService.update().setSql("total_stock = total_stock-1")
				.eq("coupon_id", seckillOrder.getCouponId())
				.gt("total_stock", 0)
				.update();//执行
		//stream group提交
		//参数：key，组名，ID
		redisTemplate.opsForStream().acknowledge(mapRecord.getStream(),"mygroup" ,mapRecord.getId());

	}

}